#!/usr/bin/env python3
# SPDX-License-Identifier: MulanPSL-2.0+
# Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.

"""
ASCII visualization module for bisect results

Provides visual representation of:
1. Functional bisect: Simple good/bad status per commit
2. Performance bisect: Metric values with linear separability visualization
3. Git tree structure: Shows branch/merge topology with test results
"""

import math
import subprocess
from typing import List, Dict, Tuple, Optional, Union, Set
from dataclasses import dataclass
from enum import Enum


class CommitStatus(Enum):
    GOOD = "good"
    BAD = "bad"
    SKIP = "skip"
    ABORT = "abort"  # For aborted jobs
    UNKNOWN = "unknown"
    SUCCESS = "success"  # For job_health status


@dataclass
class BisectPoint:
    """Single bisect test point"""
    commit: str
    value: Optional[float]  # None for functional bisect
    status: CommitStatus
    is_boundary: bool = False  # True for good_commit and bad_commit


class BisectVisualizer:
    """ASCII visualization for bisect results"""

    # Character sets for different styles
    CHARS = {
        'good': '*',
        'bad': '■',
        'skip': '?',
        'abort': 'X',  # Aborted job
        'unknown': '.',
        'boundary_good': 'G',
        'boundary_bad': 'B',
        'midpoint': '-',
        'axis': '│',
        'axis_corner': '└',
        'tick': '┴',
    }

    def __init__(self, width: int = 80, height: int = 20):
        """
        Initialize visualizer

        Args:
            width: Chart width in characters
            height: Chart height in characters
        """
        self.width = max(width, 40)
        self.height = max(height, 10)

    def visualize_functional_bisect(self, commits: List[Dict[str, str]]) -> str:
        """
        Visualize functional bisect results (good/bad only)

        Args:
            commits: List of {'commit': 'abc123', 'status': 'good'/'bad'/'skip'}

        Returns:
            ASCII visualization string
        """
        if not commits:
            return "No commits to display"

        lines = []
        lines.append("Functional Bisect Results")
        lines.append("=" * self.width)
        lines.append("")

        # Create status bar
        status_bar = []
        commit_labels = []

        for i, commit_info in enumerate(commits):
            commit = commit_info['commit'][:7]  # Short hash
            status = CommitStatus(commit_info.get('status', 'unknown'))

            # Determine character
            if i == 0:
                char = self.CHARS['boundary_good'] if status == CommitStatus.GOOD else self.CHARS[status.value]
            elif i == len(commits) - 1:
                char = self.CHARS['boundary_bad'] if status == CommitStatus.BAD else self.CHARS[status.value]
            else:
                char = self.CHARS[status.value]

            status_bar.append(char)
            commit_labels.append(commit)

        # Build the visualization
        lines.append("  " + " ".join(status_bar))
        lines.append("  " + "|" * len(status_bar))

        # Add commit labels (staggered if too many)
        if len(commits) <= 10:
            lines.append("  " + " ".join(commit_labels))
        else:
            # Stagger labels for readability
            odd_labels = [commit_labels[i] if i % 2 == 0 else "       " for i in range(len(commit_labels))]
            even_labels = [commit_labels[i] if i % 2 == 1 else "       " for i in range(len(commit_labels))]
            lines.append("  " + " ".join(odd_labels))
            lines.append("  " + " ".join(even_labels))

        lines.append("")
        lines.append("Legend: G=Good boundary, B=Bad boundary, *=good, ■=bad, ?=skip, X=abort")

        return "\n".join(lines)

    def visualize_performance_bisect(self,
                                   points: List[Dict[str, Union[str, float]]],
                                   metric_name: str,
                                   mid_point: Optional[float] = None,
                                   direction: int = 1) -> str:
        """
        Visualize performance bisect with metric values

        Args:
            points: List of {'commit': 'abc123', 'value': 95.5, 'status': 'good'/'bad'}
            metric_name: Name of the metric being measured
            mid_point: Optional threshold value for good/bad separation
            direction: 1 if higher is better, -1 if lower is better

        Returns:
            ASCII chart showing metric values with good/bad indicators
        """
        if not points:
            return "No data points to display"

        # Extract values
        values = [p['value'] for p in points]
        min_val = min(values)
        max_val = max(values)
        value_range = max_val - min_val

        if value_range == 0:
            value_range = 1  # Avoid division by zero

        # Prepare chart
        chart_width = self.width - 15  # Leave space for axis labels
        chart_height = self.height - 8  # Leave space for labels and legend

        lines = []
        lines.append(f"Performance Bisect: {metric_name}")
        lines.append("=" * self.width)
        lines.append(f"Direction: {'Higher is better' if direction == 1 else 'Lower is better'}")
        if mid_point is not None:
            lines.append(f"Mid-point threshold: {mid_point:.2f}")
        lines.append("")

        # Create the chart grid
        chart = [[' ' for _ in range(chart_width)] for _ in range(chart_height)]

        # Plot points
        for i, point in enumerate(points):
            x = int((i / max(len(points) - 1, 1)) * (chart_width - 1))
            y = chart_height - 1 - int(((point['value'] - min_val) / value_range) * (chart_height - 1))

            status = CommitStatus(point.get('status', 'unknown'))

            # Choose character based on status
            if i == 0 and status == CommitStatus.GOOD:
                char = self.CHARS['boundary_good']
            elif i == len(points) - 1 and status == CommitStatus.BAD:
                char = self.CHARS['boundary_bad']
            else:
                char = self.CHARS[status.value]

            if 0 <= x < chart_width and 0 <= y < chart_height:
                chart[y][x] = char

        # Add mid_point line if provided
        if mid_point is not None and min_val <= mid_point <= max_val:
            mid_y = chart_height - 1 - int(((mid_point - min_val) / value_range) * (chart_height - 1))
            if 0 <= mid_y < chart_height:
                for x in range(chart_width):
                    if chart[mid_y][x] == ' ':
                        chart[mid_y][x] = self.CHARS['midpoint']

        # Build the output with axis
        for y in range(chart_height):
            # Add Y-axis label
            if y == 0:
                label = f"{max_val:8.2f}"
            elif y == chart_height - 1:
                label = f"{min_val:8.2f}"
            elif y == chart_height // 2:
                label = f"{(max_val + min_val) / 2:8.2f}"
            else:
                label = "        "

            line = f"{label} {self.CHARS['axis']} {''.join(chart[y])}"
            lines.append(line)

        # Add X-axis
        lines.append(" " * 9 + self.CHARS['axis_corner'] + "─" * chart_width)

        # Add commit labels (abbreviated)
        if len(points) <= 10:
            commit_labels = []
            for i, point in enumerate(points):
                x = int((i / max(len(points) - 1, 1)) * (chart_width - 1))
                commit = point['commit'][:7]
                commit_labels.append((x, commit))

            # Build label line
            label_line = [' '] * (chart_width + 10)
            for x, commit in commit_labels:
                for j, c in enumerate(commit):
                    if x + j + 10 < len(label_line):
                        label_line[x + j + 10] = c
            lines.append(''.join(label_line))
        else:
            lines.append(" " * 10 + f"[{points[0]['commit'][:7]} ... {points[-1]['commit'][:7]}]")

        # Add legend
        lines.append("")
        lines.append("Legend: G=Good boundary, B=Bad boundary, *=good, ■=bad, ?=skip, X=abort")
        if mid_point is not None:
            lines.append(f"        {self.CHARS['midpoint']}=mid-point threshold ({mid_point:.2f})")

        # Add statistics
        lines.append("")
        lines.append("Statistics:")
        good_points = [p for p in points if p.get('status') == 'good']
        bad_points = [p for p in points if p.get('status') == 'bad']

        if good_points:
            good_values = [p['value'] for p in good_points]
            lines.append(f"  Good samples: {len(good_points)}, range: [{min(good_values):.2f}, {max(good_values):.2f}]")

        if bad_points:
            bad_values = [p['value'] for p in bad_points]
            lines.append(f"  Bad samples:  {len(bad_points)}, range: [{min(bad_values):.2f}, {max(bad_values):.2f}]")

        return "\n".join(lines)

    def visualize_bisect_progression(self,
                                    tested_commits: List[Dict],
                                    all_commits: List[str],
                                    current_commit: Optional[str] = None) -> str:
        """
        Visualize bisect progression showing tested vs untested commits

        Args:
            tested_commits: List of tested commits with status
            all_commits: List of all commits in range
            current_commit: Currently testing commit

        Returns:
            ASCII visualization of bisect progress
        """
        lines = []
        lines.append("Bisect Progress")
        lines.append("=" * self.width)

        # Create status map
        status_map = {c['commit']: c['status'] for c in tested_commits}

        # Progress bar
        tested_count = len(tested_commits)
        total_count = len(all_commits)
        progress_pct = (tested_count / total_count) * 100 if total_count > 0 else 0

        bar_width = self.width - 20
        filled = int((tested_count / total_count) * bar_width) if total_count > 0 else 0

        progress_bar = "[" + "=" * filled + ">" + " " * (bar_width - filled - 1) + "]"
        lines.append(f"Progress: {progress_bar} {progress_pct:.1f}%")
        lines.append(f"Tested: {tested_count}/{total_count} commits")
        lines.append("")

        # Commit timeline
        if len(all_commits) <= 50:
            timeline = []
            for commit in all_commits:
                if commit == current_commit:
                    timeline.append("@")  # Currently testing
                elif commit in status_map:
                    status = status_map[commit]
                    if status == "good":
                        timeline.append("*")
                    elif status == "bad":
                        timeline.append("■")
                    elif status == "skip":
                        timeline.append("?")
                    elif status == "abort":
                        timeline.append("X")
                    else:
                        timeline.append("?")  # Unknown or other status
                else:
                    timeline.append(".")  # Not tested yet

            # Display timeline in rows
            row_length = self.width - 10
            for i in range(0, len(timeline), row_length):
                row = timeline[i:i+row_length]
                lines.append("  " + "".join(row))
        else:
            lines.append(f"  [Timeline too long to display: {len(all_commits)} commits]")

        lines.append("")
        lines.append("Legend: @=current, *=good, ■=bad, ?=skip, X=abort, .=untested")

        return "\n".join(lines)

    def visualize_git_tree(self,
                          repo_path: str,
                          good_commit: str,
                          bad_commit: str,
                          tested_commits: Dict[str, str],
                          max_commits: int = 50) -> str:
        """
        Visualize git commit tree structure with bisect test results

        Args:
            repo_path: Path to git repository
            good_commit: Good boundary commit
            bad_commit: Bad boundary commit
            tested_commits: Dict of {commit_hash: status} for tested commits
            max_commits: Maximum number of commits to display

        Returns:
            ASCII tree visualization with test status markers
        """
        try:
            # Get git log with graph structure
            cmd = [
                'git', '-C', repo_path, 'log', '--graph', '--oneline',
                '--ancestry-path', f'{good_commit}..{bad_commit}',
                f'--max-count={max_commits}', '--decorate'
            ]

            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            git_graph = result.stdout

            if not git_graph:
                return "No commits found in range"

            # Process each line to add status markers
            lines = []
            lines.append("Git Commit Tree with Bisect Results")
            lines.append("=" * self.width)
            lines.append("")

            for line in git_graph.split('\n'):
                if not line:
                    continue

                # Extract commit hash from the line
                # Git log format: "* abc1234 Commit message"
                commit_hash = None
                parts = line.split()
                for part in parts:
                    if len(part) >= 7 and all(c in '0123456789abcdef' for c in part[:7]):
                        commit_hash = part[:7]
                        break

                # Add status marker if commit was tested
                if commit_hash:
                    full_hash = self._get_full_hash(repo_path, commit_hash)
                    if full_hash in tested_commits:
                        status = tested_commits[full_hash]
                        marker = self._get_tree_marker(status)
                        # Replace commit hash with marked version
                        line = line.replace(commit_hash, f"{marker} {commit_hash}")
                    elif full_hash == good_commit[:7] or full_hash == good_commit:
                        line = line.replace(commit_hash, f"[G] {commit_hash}")
                    elif full_hash == bad_commit[:7] or full_hash == bad_commit:
                        line = line.replace(commit_hash, f"[B] {commit_hash}")

                lines.append(line)

            lines.append("")
            lines.append("Legend: [G]=Good boundary, [B]=Bad boundary")
            lines.append("        [+]=tested good, [X]=tested bad, [?]=skip, [A]=abort, [ ]=untested")

            # Add commit count summary
            total_in_range = self._count_commits_in_range(repo_path, good_commit, bad_commit)
            tested_count = len(tested_commits)
            lines.append("")
            lines.append(f"Commits in range: {total_in_range}, Tested: {tested_count}")

            return "\n".join(lines)

        except subprocess.CalledProcessError as e:
            return f"Failed to get git tree: {e}"
        except Exception as e:
            return f"Error visualizing git tree: {e}"

    def visualize_merge_analysis(self,
                                repo_path: str,
                                first_bad_commit: str,
                                tested_commits: Dict[str, str]) -> str:
        """
        Analyze and visualize merge commits in the bisect path

        Args:
            repo_path: Path to git repository
            first_bad_commit: The identified first bad commit
            tested_commits: Dict of tested commits and their status

        Returns:
            ASCII visualization focusing on merge commits
        """
        lines = []
        lines.append("Merge Commit Analysis")
        lines.append("=" * self.width)

        try:
            # Find merge commits in tested set
            merge_commits = []
            for commit, status in tested_commits.items():
                if self._is_merge_commit(repo_path, commit):
                    merge_commits.append((commit, status))

            if not merge_commits:
                lines.append("No merge commits found in bisect path")
                return "\n".join(lines)

            lines.append(f"Found {len(merge_commits)} merge commits in bisect path:")
            lines.append("")

            # Analyze each merge commit
            for commit, status in merge_commits:
                parents = self._get_commit_parents(repo_path, commit)
                commit_msg = self._get_commit_message(repo_path, commit)

                marker = self._get_tree_marker(status)
                lines.append(f"{marker} {commit[:7]}: {commit_msg[:50]}")

                if len(parents) >= 2:
                    # Check which parent branch contains the regression
                    parent_status = []
                    for parent in parents:
                        if parent in tested_commits:
                            parent_status.append(f"{parent[:7]} ({tested_commits[parent]})")
                        else:
                            parent_status.append(f"{parent[:7]} (untested)")

                    lines.append(f"    Parents: {', '.join(parent_status)}")

                    # If this is the first bad commit and it's a merge
                    if commit == first_bad_commit:
                        lines.append("    -> This merge introduced the regression")
                        # Try to identify which branch
                        bad_parents = [p for p in parents if p in tested_commits and tested_commits[p] == 'bad']
                        good_parents = [p for p in parents if p in tested_commits and tested_commits[p] == 'good']

                        if good_parents and not bad_parents:
                            lines.append("    -> The merge itself caused the regression")
                        elif bad_parents and not good_parents:
                            lines.append(f"    -> Regression came from parent: {bad_parents[0][:7]}")

                lines.append("")

            return "\n".join(lines)

        except Exception as e:
            lines.append(f"Error analyzing merges: {e}")
            return "\n".join(lines)

    def _get_tree_marker(self, status: str) -> str:
        """Get tree marker for commit status"""
        markers = {
            'good': '[+]',
            'success': '[+]',  # Treat success same as good
            'bad': '[X]',
            'skip': '[?]',
            'abort': '[A]',  # Aborted job
            'unknown': '[ ]'
        }
        return markers.get(status, '[ ]')

    def _get_full_hash(self, repo_path: str, short_hash: str) -> str:
        """Get full commit hash from short hash"""
        try:
            cmd = ['git', '-C', repo_path, 'rev-parse', short_hash]
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            return result.stdout.strip()
        except:
            return short_hash

    def _is_merge_commit(self, repo_path: str, commit: str) -> bool:
        """Check if commit is a merge commit"""
        try:
            cmd = ['git', '-C', repo_path, 'rev-list', '--parents', '-n', '1', commit]
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            # Merge commits have more than one parent
            return len(result.stdout.strip().split()) > 2
        except:
            return False

    def _get_commit_parents(self, repo_path: str, commit: str) -> List[str]:
        """Get parent commits"""
        try:
            cmd = ['git', '-C', repo_path, 'rev-list', '--parents', '-n', '1', commit]
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            parts = result.stdout.strip().split()
            return parts[1:] if len(parts) > 1 else []
        except:
            return []

    def _get_commit_message(self, repo_path: str, commit: str) -> str:
        """Get commit message"""
        try:
            cmd = ['git', '-C', repo_path, 'log', '--format=%s', '-n', '1', commit]
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            return result.stdout.strip()
        except:
            return "Unknown"

    def _count_commits_in_range(self, repo_path: str, start: str, end: str) -> int:
        """Count commits in range"""
        try:
            cmd = ['git', '-C', repo_path, 'rev-list', '--count', f'{start}..{end}']
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            return int(result.stdout.strip())
        except:
            return 0


def demo():
    """Demo of bisect visualizations"""
    visualizer = BisectVisualizer(width=80, height=15)

    # Demo 1: Functional bisect
    print("\n" + "=" * 80)
    print("DEMO 1: Functional Bisect")
    print("=" * 80)

    functional_commits = [
        {'commit': 'abc1234', 'status': 'good'},
        {'commit': 'def5678', 'status': 'good'},
        {'commit': 'ghi9012', 'status': 'skip'},
        {'commit': 'jkl3456', 'status': 'bad'},
        {'commit': 'mno7890', 'status': 'bad'},
    ]

    print(visualizer.visualize_functional_bisect(functional_commits))

    # Demo 2: Performance bisect
    print("\n" + "=" * 80)
    print("DEMO 2: Performance Bisect")
    print("=" * 80)

    performance_points = [
        {'commit': 'v5.0', 'value': 100.0, 'status': 'good'},
        {'commit': 'abc1234', 'value': 98.5, 'status': 'good'},
        {'commit': 'def5678', 'value': 95.2, 'status': 'good'},
        {'commit': 'ghi9012', 'value': 89.7, 'status': 'bad'},
        {'commit': 'jkl3456', 'value': 85.3, 'status': 'bad'},
        {'commit': 'v5.1', 'value': 82.1, 'status': 'bad'},
    ]

    print(visualizer.visualize_performance_bisect(
        performance_points,
        metric_name="throughput_mb_per_sec",
        mid_point=92.5,
        direction=1
    ))

    # Demo 3: Bisect progression
    print("\n" + "=" * 80)
    print("DEMO 3: Bisect Progression")
    print("=" * 80)

    all_commits = [f"commit_{i:03d}" for i in range(20)]
    tested = [
        {'commit': all_commits[0], 'status': 'good'},
        {'commit': all_commits[19], 'status': 'bad'},
        {'commit': all_commits[10], 'status': 'bad'},
        {'commit': all_commits[5], 'status': 'good'},
        {'commit': all_commits[7], 'status': 'skip'},
    ]

    print(visualizer.visualize_bisect_progression(
        tested_commits=tested,
        all_commits=all_commits,
        current_commit=all_commits[8]
    ))


if __name__ == "__main__":
    demo()
